package org.apache.hadoop.hive.cassandra.output.cql;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Properties;

import org.apache.hadoop.hive.cassandra.serde.AbstractCassandraSerDe;
import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.util.Progressable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.Metadata;
import com.datastax.driver.core.Session;

public class CqlRecordWriter implements FileSinkOperator.RecordWriter {
	
	private final static Logger LOGGER  = LoggerFactory.getLogger(CqlRecordWriter.class);
	
	private AbstractCqlMapper cqlMapper;
	
	//private CassandraProxyClient client;
	
	private Cluster cluster;
	private Session session;
	
	public CqlRecordWriter(JobConf jobConf, Properties properties, Progressable progressable) throws IOException {
        final String cassandraHost = jobConf.get(AbstractCassandraSerDe.CASSANDRA_HOST);
        final String cassandraKeySpace = jobConf.get(AbstractCassandraSerDe.CASSANDRA_KEYSPACE_NAME);
        //final int cassandraPort =  Integer.parseInt(jobConf.get(AbstractCassandraSerDe.CASSANDRA_PORT));
        /**
        try {
            client = new CassandraProxyClient(cassandraHost, cassandraPort, true, true);
        } catch (CassandraException e) {
            throw new IOException(e);
        }
        **/
        String[] hosts = cassandraHost.split(",");
        cluster = Cluster.builder().addContactPoints(hosts).build();
        
        Metadata metadata = cluster.getMetadata();
        LOGGER.info("Connected to cluster:{}", metadata.getClusterName());
		for (Host host : metadata.getAllHosts()) {
			LOGGER.info("Datatacenter: {}; Host: {}; Rack: {}", host.getDatacenter(), host.getAddress(),host.getRack());
		}
		session = cluster.connect(cassandraKeySpace);
        //cqlMapper = new ThriftServerCqlMapper(jobConf,client);
		cqlMapper = new DatastaxCqlMapMapper(session, jobConf);
	}
	

	@Override
	public void write(Writable w) throws IOException {
		cqlMapper.writeToCassandra( w);
	}

	@Override
	public void close(boolean abort) throws IOException {
		cqlMapper.finishWriteJob();
		//client.close();
		session.close();
		cluster.close();
	}
}
